package com.niit.realtime.kafka;


import com.niit.realtime.service.StatsService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class KafkaConsumer {

    private final StatsService statsService;

    public KafkaConsumer(StatsService statsService) {
        this.statsService = statsService;
    }

    @KafkaListener(topics = "order_valid_invalid_count")
    public void consumeOrderValidInvalidCount(String message) {
        try {
            log.info("Received message from order_valid_invalid_count: {}", message);
            String[] parts = message.split("\t");
            if (parts.length == 2) {
                String isValid = parts[0].trim();
                int count = Integer.parseInt(parts[1].trim());
                statsService.processOrderValidInvalidCount(isValid, count);
            }
        } catch (Exception e) {
            log.error("Error processing order_valid_invalid_count message: {}", message, e);
        }
    }

    @KafkaListener(topics = "product_order_stats")
    public void consumeProductOrderStats(String message) {
        try {
            log.info("Received message from product_order_stats: {}", message);
            String[] parts = message.split("\t");
            if (parts.length == 3) {
                String productName = parts[0].trim();
                String isValid = parts[1].trim();
                int count = Integer.parseInt(parts[2].trim());
                statsService.processProductOrderStats(productName, isValid, count);
            }
        } catch (Exception e) {
            log.error("Error processing product_order_stats message: {}", message, e);
        }
    }

    @KafkaListener(topics = "category_order_counts")
    public void consumeCategoryOrderCounts(String message) {
        try {
            log.info("Received message from category_order_counts: {}", message);
            String[] parts = message.split("\t");
            if (parts.length == 2) {
                String category = parts[0].trim();
                int count = Integer.parseInt(parts[1].trim());
                statsService.processCategoryOrderCounts(category, count);
            }
        } catch (Exception e) {
            log.error("Error processing category_order_counts message: {}", message, e);
        }
    }

    @KafkaListener(topics = "category_valid_invalid_counts")
    public void consumeCategoryValidInvalidCounts(String message) {
        try {
            log.info("Received message from category_valid_invalid_counts: {}", message);
            // 处理逻辑类似，根据业务需求实现
        } catch (Exception e) {
            log.error("Error processing category_valid_invalid_counts message: {}", message, e);
        }
    }

    @KafkaListener(topics = "category_product_order_stats")
    public void consumeCategoryProductOrderStats(String message) {
        try {
            log.info("Received message from category_product_order_stats: {}", message);
            String[] parts = message.split("\t");
            if (parts.length == 4) {
                String category = parts[0].trim();
                String productName = parts[1].trim();
                String isValid = parts[2].trim();
                int count = Integer.parseInt(parts[3].trim());
                statsService.processCategoryProductOrderStats(category, productName, isValid, count);
            }
        } catch (Exception e) {
            log.error("Error processing category_product_order_stats message: {}", message, e);
        }
    }
}